package com.datagateway.component;

import com.datagateway.config.DataGatewayProperties;
import com.datagateway.model.ProcessedData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 数据分区管理器
 * 负责管理数据的分区策略和分区信息
 * 
 * @author Data Gateway Team
 * @version 1.0.0
 */
@Component
public class PartitionManager {

    private static final Logger logger = LoggerFactory.getLogger(PartitionManager.class);

    @Autowired
    private DataGatewayProperties properties;

    /**
     * 分区策略枚举
     */
    public enum PartitionStrategy {
        DATE,           // 按日期分区
        BUSINESS,       // 按业务类型分区
        HASH,           // 按哈希值分区
        CUSTOM          // 自定义分区
    }

    /**
     * 分区信息缓存
     */
    private final Map<String, PartitionInfo> partitionCache = new ConcurrentHashMap<>();

    /**
     * 分区统计信息
     */
    private final AtomicLong totalPartitions = new AtomicLong(0);
    private final AtomicLong activePartitions = new AtomicLong(0);

    /**
     * 默认日期格式
     */
    private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");

    /**
     * 获取数据的分区信息
     * 
     * @param data 处理后的数据
     * @return 分区信息
     */
    public PartitionInfo getPartitionInfo(ProcessedData data) {
        try {
            // 根据配置的分区策略获取分区信息
            PartitionStrategy strategy = getPartitionStrategy();
            
            switch (strategy) {
                case DATE:
                    return getDatePartition(data);
                case BUSINESS:
                    return getBusinessPartition(data);
                case HASH:
                    return getHashPartition(data);
                case CUSTOM:
                    return getCustomPartition(data);
                default:
                    return getDefaultPartition(data);
            }
        } catch (Exception e) {
            logger.error("获取分区信息失败: dataId={}", data.getId(), e);
            return getDefaultPartition(data);
        }
    }

    /**
     * 按日期分区
     * 
     * @param data 数据
     * @return 分区信息
     */
    private PartitionInfo getDatePartition(ProcessedData data) {
        LocalDate partitionDate;
        
        // 优先使用数据时间戳
        if (data.getDataTimestamp() != null) {
            partitionDate = data.getDataTimestamp().toLocalDate();
        } else if (data.getProcessTime() != null) {
            partitionDate = data.getProcessTime().toLocalDate();
        } else {
            partitionDate = LocalDate.now();
        }
        
        String partitionValue = partitionDate.format(DATE_FORMATTER);
        String partitionKey = "dt";
        
        return createPartitionInfo(partitionKey, partitionValue, "date", partitionDate);
    }

    /**
     * 按业务类型分区
     * 
     * @param data 数据
     * @return 分区信息
     */
    private PartitionInfo getBusinessPartition(ProcessedData data) {
        String businessType = extractBusinessType(data);
        String partitionKey = "business_type";
        
        return createPartitionInfo(partitionKey, businessType, "business", null);
    }

    /**
     * 按哈希值分区
     * 
     * @param data 数据
     * @return 分区信息
     */
    private PartitionInfo getHashPartition(ProcessedData data) {
        int hashValue = Math.abs(data.getId().hashCode()) % 10; // 分为10个分区
        String partitionValue = String.format("p%02d", hashValue);
        String partitionKey = "hash_partition";
        
        return createPartitionInfo(partitionKey, partitionValue, "hash", null);
    }

    /**
     * 自定义分区
     * 
     * @param data 数据
     * @return 分区信息
     */
    private PartitionInfo getCustomPartition(ProcessedData data) {
        // 这里可以根据具体业务需求实现自定义分区逻辑
        // 例如：按用户ID、地区、产品类型等分区
        
        String customValue = extractCustomPartitionValue(data);
        String partitionKey = "custom_partition";
        
        return createPartitionInfo(partitionKey, customValue, "custom", null);
    }

    /**
     * 默认分区
     * 
     * @param data 数据
     * @return 分区信息
     */
    private PartitionInfo getDefaultPartition(ProcessedData data) {
        LocalDate today = LocalDate.now();
        String partitionValue = today.format(DATE_FORMATTER);
        String partitionKey = "dt";
        
        return createPartitionInfo(partitionKey, partitionValue, "default", today);
    }

    /**
     * 创建分区信息
     * 
     * @param partitionKey 分区键
     * @param partitionValue 分区值
     * @param partitionType 分区类型
     * @param partitionDate 分区日期
     * @return 分区信息
     */
    private PartitionInfo createPartitionInfo(String partitionKey, String partitionValue, 
                                            String partitionType, LocalDate partitionDate) {
        String partitionId = partitionKey + "=" + partitionValue;
        
        PartitionInfo partitionInfo = partitionCache.computeIfAbsent(partitionId, 
            k -> new PartitionInfo(partitionKey, partitionValue, partitionType, partitionDate));
        
        // 更新统计信息
        totalPartitions.incrementAndGet();
        if (partitionInfo.isActive()) {
            activePartitions.incrementAndGet();
        }
        
        return partitionInfo;
    }

    /**
     * 从数据中提取业务类型
     * 
     * @param data 数据
     * @return 业务类型
     */
    private String extractBusinessType(ProcessedData data) {
        try {
            if (data.getTransformedData() != null) {
                // 尝试从转换后的数据中提取业务类型
                Object businessType = data.getTransformedData().get("business_type");
                if (businessType != null) {
                    return businessType.toString();
                }
                
                // 尝试从其他字段推断业务类型
                Object eventType = data.getTransformedData().get("event_type");
                if (eventType != null) {
                    return eventType.toString();
                }
            }
            
            // 从原始数据中提取
            if (data.getOriginalData() != null && data.getOriginalData().contains("business_type")) {
                // 简单的JSON解析，实际项目中可以使用更robust的解析方法
                String[] parts = data.getOriginalData().split("\"business_type\"");
                if (parts.length > 1) {
                    String value = parts[1].split("\"")[2];
                    return value;
                }
            }
            
        } catch (Exception e) {
            logger.warn("提取业务类型失败: dataId={}", data.getId(), e);
        }
        
        return "default";
    }

    /**
     * 从数据中提取自定义分区值
     * 
     * @param data 数据
     * @return 自定义分区值
     */
    private String extractCustomPartitionValue(ProcessedData data) {
        try {
            if (data.getTransformedData() != null) {
                // 可以根据具体业务需求实现
                Object userId = data.getTransformedData().get("user_id");
                if (userId != null) {
                    return "user_" + (Math.abs(userId.hashCode()) % 100);
                }
                
                Object region = data.getTransformedData().get("region");
                if (region != null) {
                    return region.toString();
                }
            }
        } catch (Exception e) {
            logger.warn("提取自定义分区值失败: dataId={}", data.getId(), e);
        }
        
        return "default";
    }

    /**
     * 获取分区策略
     * 
     * @return 分区策略
     */
    private PartitionStrategy getPartitionStrategy() {
        // 这里可以从配置中读取分区策略
        // 暂时返回默认的日期分区策略
        return PartitionStrategy.DATE;
    }

    /**
     * 获取所有活跃分区
     * 
     * @return 活跃分区列表
     */
    public List<PartitionInfo> getActivePartitions() {
        return partitionCache.values().stream()
            .filter(PartitionInfo::isActive)
            .sorted(Comparator.comparing(PartitionInfo::getPartitionValue))
            .collect(java.util.stream.Collectors.toList());
    }

    /**
     * 获取分区统计信息
     * 
     * @return 分区统计信息
     */
    public PartitionStatistics getPartitionStatistics() {
        return new PartitionStatistics(
            totalPartitions.get(),
            activePartitions.get(),
            partitionCache.size(),
            getActivePartitions().size()
        );
    }

    /**
     * 清理过期分区
     * 
     * @param retentionDays 保留天数
     */
    public void cleanupExpiredPartitions(int retentionDays) {
        LocalDate cutoffDate = LocalDate.now().minusDays(retentionDays);
        
        partitionCache.entrySet().removeIf(entry -> {
            PartitionInfo partition = entry.getValue();
            if (partition.getPartitionDate() != null && 
                partition.getPartitionDate().isBefore(cutoffDate)) {
                logger.info("清理过期分区: {}", entry.getKey());
                return true;
            }
            return false;
        });
    }

    /**
     * 分区信息类
     */
    public static class PartitionInfo {
        private final String partitionKey;
        private final String partitionValue;
        private final String partitionType;
        private final LocalDate partitionDate;
        private final LocalDateTime createTime;
        private volatile boolean active;
        private final AtomicLong recordCount = new AtomicLong(0);

        public PartitionInfo(String partitionKey, String partitionValue, String partitionType, LocalDate partitionDate) {
            this.partitionKey = partitionKey;
            this.partitionValue = partitionValue;
            this.partitionType = partitionType;
            this.partitionDate = partitionDate;
            this.createTime = LocalDateTime.now();
            this.active = true;
        }

        // Getter和Setter方法
        public String getPartitionKey() { return partitionKey; }
        public String getPartitionValue() { return partitionValue; }
        public String getPartitionType() { return partitionType; }
        public LocalDate getPartitionDate() { return partitionDate; }
        public LocalDateTime getCreateTime() { return createTime; }
        public boolean isActive() { return active; }
        public void setActive(boolean active) { this.active = active; }
        public long getRecordCount() { return recordCount.get(); }
        public void incrementRecordCount() { recordCount.incrementAndGet(); }

        public String getPartitionId() {
            return partitionKey + "=" + partitionValue;
        }

        @Override
        public String toString() {
            return String.format("PartitionInfo{key=%s, value=%s, type=%s, date=%s, active=%s, records=%d}",
                               partitionKey, partitionValue, partitionType, partitionDate, active, recordCount.get());
        }
    }

    /**
     * 分区统计信息类
     */
    public static class PartitionStatistics {
        private final long totalPartitions;
        private final long activePartitions;
        private final int cachedPartitions;
        private final int currentActivePartitions;

        public PartitionStatistics(long totalPartitions, long activePartitions, 
                                 int cachedPartitions, int currentActivePartitions) {
            this.totalPartitions = totalPartitions;
            this.activePartitions = activePartitions;
            this.cachedPartitions = cachedPartitions;
            this.currentActivePartitions = currentActivePartitions;
        }

        // Getter方法
        public long getTotalPartitions() { return totalPartitions; }
        public long getActivePartitions() { return activePartitions; }
        public int getCachedPartitions() { return cachedPartitions; }
        public int getCurrentActivePartitions() { return currentActivePartitions; }

        @Override
        public String toString() {
            return String.format("PartitionStatistics{totalPartitions=%d, activePartitions=%d, " +
                               "cachedPartitions=%d, currentActivePartitions=%d}",
                               totalPartitions, activePartitions, cachedPartitions, currentActivePartitions);
        }
    }
}
